package com.opc.websocket;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;

@Component
@Slf4j
@ServerEndpoint("/ws/{sid}")
public class WebSocketServer {

    // 使用ConcurrentHashMap保证线程安全
    private static final Map<String, Session> sessionMap = new ConcurrentHashMap<>();

    // 使用阻塞队列存储待发送的消息
    private final BlockingQueue<MessageTask> messageQueue = new LinkedBlockingQueue<>();

    // 单线程执行器处理消息发送
    private final ExecutorService senderExecutor = Executors.newSingleThreadExecutor();

    public WebSocketServer() {
        // 启动消息处理线程
        senderExecutor.submit(this::messageProcessingLoop);
    }

    // 消息任务类，封装要发送的消息和目标会话
    private static class MessageTask {
        private final Session session;
        private final String message;

        public MessageTask(Session session, String message) {
            this.session = session;
            this.message = message;
        }

        public Session getSession() {
            return session;
        }

        public String getMessage() {
            return message;
        }
    }

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("sid") String sid) {
        System.out.println("客户端建立连接, ID：" + sid);
        sessionMap.put(sid, session);

        try {
            session.getBasicRemote().sendText("连接成功，你的ID是：" + sid);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 收到客户端消息后调用的方法
     */
    @OnMessage
    public void onMessage(String message, @PathParam("sid") String sid) {
        System.out.println("收到来自客户端：" + sid + "的信息:" + message);
        if ("close-jar".equals(message)) {
            System.out.println("关闭java");
            // 优雅关闭
            closeResources();
            System.exit(0);
        }
    }

    private void closeJarProcess() {
        // 原有方法保持不变
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(@PathParam("sid") String sid) {
        System.out.println("连接断开:" + sid);
        sessionMap.remove(sid);
    }

    /**
     * 错误处理
     */
    @OnError
    public void onError(Throwable error) {
        error.printStackTrace();
    }

    /**
     * 群发消息
     */
//    public void sendToAllClient(String message) {
//        // 复制当前会话集合，避免并发修改问题
//        Collection<Session> sessions = new ArrayList<>(sessionMap.values());
//
//        for (Session session : sessions) {
//            if (session != null && session.isOpen()) {
//                // 将消息加入队列，由单独线程处理发送
//                messageQueue.offer(new MessageTask(session, message));
//            }
//        }
//    }

    // 线程池监控示例（可定期打印或接入监控系统）
    public void monitorThreadPool() {
        ThreadPoolExecutor executor = (ThreadPoolExecutor) sendExecutor;
        log.info("线程池状态：活跃线程数={}, 队列任务数={}, 已完成任务数={}",
                executor.getActiveCount(),
                executor.getQueue().size(),
                executor.getCompletedTaskCount());
    }

    //解决发多了卡顿？？？的问题？？？

    // 初始化线程池（核心线程数根据CPU核心数和业务调整）
    // 将单线程改为线程池，并行处理多个发送任务，提高消费速度
    private ExecutorService sendExecutor = new ThreadPoolExecutor(
            8, // 核心线程数
            16, // 最大线程数
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000), // 任务队列（限制大小，避免OOM）
            new ThreadPoolExecutor.CallerRunsPolicy() // 队列满时，让提交任务的线程帮忙处理（限流）
    );

    // 假设sessionMap已定义为线程安全的ConcurrentHashMap
//    private final ConcurrentHashMap<String, Session> sessionMap = new ConcurrentHashMap<>();

    //版本1
//    public void sendToAllClient(String message) {
//        // 直接遍历ConcurrentHashMap的values（支持并发修改，遍历安全）
//        for (Session session : sessionMap.values()) {
//            // 1. 过滤无效会话（先判断null，再判断是否打开）
//            if (session == null) {
//                continue; // 跳过null值（理论上ConcurrentHashMap不会存null，但防御性判断）
//            }
//            if (!session.isOpen()) {
//                // 清理已关闭的会话（根据session.getId()移除，避免空指针）
//                sessionMap.remove(session.getId());
//                continue;
//            }
//
//            // 2. 提交发送任务到线程池
//            sendExecutor.submit(() -> {
//                try {
//                    // 使用异步发送并添加结果回调（确保知道发送是否成功）
//                    session.getAsyncRemote().sendText(message, result -> {
//                        if (!result.isOK()) {
//                            log.error("会话[{}]消息发送失败", session.getId(), result.getException());
//                            // 发送失败时清理无效会话
//                            sessionMap.remove(session.getId());
//                        }
//                    });
//                } catch (Exception e) {
//                    log.error("会话[{}]发送任务提交失败", session.getId(), e);
//                    sessionMap.remove(session.getId());
//                }
//            });
//        }
//    }

    //版本2
    public void sendToAllClient(String message) {
        // 遍历 ConcurrentHashMap 的会话（确保 sessionMap 是 ConcurrentHashMap 类型）
        for (Session session : sessionMap.values()) {
            // 1. 过滤无效会话
            if (session == null) {
                continue;
            }
            if (!session.isOpen()) {
                sessionMap.remove(session.getId());
                continue;
            }

            // 关键：定义 final 局部变量，固化当前 session 引用
            final Session currentSession = session;

            // 2. 提交发送任务到线程池
            sendExecutor.submit(() -> {
                try {
                    currentSession.getAsyncRemote().sendText(message, result -> {
                        if (!result.isOK()) {
                            log.error("异步发送失败，会话ID={}", currentSession.getId(), result.getException());
                            retrySend(currentSession, message, 1);
                        }
                    });
                    // 此处使用 final 修饰的 currentSession，而非循环变量 session
//                    currentSession.getAsyncRemote().sendText(message, new SendHandler() {
//                        @Override
//                        public void onResult(SendResult result) {
//                            if (!result.isOK()) {
//                                // 回调中访问 currentSession（已固化，可安全访问）
//                                log.error("异步发送失败，会话ID={}", currentSession.getId(), result.getException());
//                                // 重试逻辑（传入固化的 currentSession）
//                                retrySend(currentSession, message, 1);
//                            }
//                        }
//                    });
                } catch (Exception e) {
                    log.error("会话[{}]发送任务提交失败", currentSession.getId(), e);
                    sessionMap.remove(currentSession.getId());
                }
            });
        }
    }


//    public void sendToAllClient(String message) {
//        // 复制会话集合（尽量减少复制次数，或用并发容器替代）
//        List<Session> sessions = new ArrayList<>(sessionMap.values());
//
//        for (Session session : sessions) {
//            if (session != null && session.isOpen()) {
//                // 实时判断：跳过 null 或已关闭的会话，减少无效任务提交
//                // 提交任务到线程池并行处理
//                sendExecutor.submit(() -> {
//                    try {
//                        // 发送消息（使用非阻塞方法，如异步发送）
//                        session.getAsyncRemote().sendText(message);
//                    } catch (Exception e) {
//                        // 处理异常（如客户端已断开）
//                        log.error("发送消息失败", e);
//                        try {
//                            session.close(); // 关闭无效会话
//                        } catch (IOException ex) {
//                            // 忽略关闭异常
//                        }
//                    }
//                });
//            }
//        }
//
//    }

    // 消息可靠性增强（解决 “丢失风险”）
    //异步发送结果回调：使用 getAsyncRemote().sendText(message, resultHandler)
    // 监听发送结果，确保失败时可重试：


    // 重试方法
    private void retrySend(Session session, String message, int retryCount) {
        if (retryCount > 3 || !session.isOpen()) { // 最多重试3次
            return;
        }
        sendExecutor.submit(() -> {
            try {
                Thread.sleep(100L * retryCount); // 指数退避
                session.getAsyncRemote().sendText(message);
            } catch (Exception e) {
                retrySend(session, message, retryCount + 1);
            }
        });
    }


    //---------------------------------

    /**
     * 消息处理循环
     */
    private void messageProcessingLoop() {
        try {
            while (true) {
                // 从队列中取出消息（阻塞等待）
                MessageTask task = messageQueue.take();
                Session session = task.getSession();
                String message = task.getMessage();

                // 再次检查会话状态
                if (session.isOpen()) {
                    try {
                        // 使用同步发送方法
                        session.getBasicRemote().sendText(message);
                    } catch (IOException e) {
                        // 处理发送异常
                        System.err.println("发送消息失败: " + e.getMessage());
                        // 关闭会话
                        closeSessionQuietly(session);
                    }
                } else {
                    // 会话已关闭，从map中移除
                    sessionMap.values().remove(session);
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            // 清理资源
            senderExecutor.shutdown();
        }
    }

    private void closeSessionQuietly(Session session) {
        try {
            if (session != null && session.isOpen()) {
                session.close();
            }
        } catch (IOException e) {
            // 忽略异常
        }
        // 从map中移除
        sessionMap.values().remove(session);
    }

    private void closeResources() {
        // 关闭线程池
        senderExecutor.shutdown();
        try {
            if (!senderExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                senderExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            senderExecutor.shutdownNow();
        }

        // 关闭所有会话
        for (Session session : sessionMap.values()) {
            closeSessionQuietly(session);
        }
        sessionMap.clear();
    }
}
